package com.david.trade.clear.component;

import com.david.trade.clear.service.TcpMessage;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author wangjijiang
 * @date 2020/6/5 11:13
 */
public class RingBufferComponent {

    public static RingBufferComponent ringBufferComponent;

    private Disruptor<TcpMessageEvent> disruptor;


    private RingBuffer<TcpMessageEvent> ringBuffer;

    private TcpMessageEventTranslator tcpMessageEventTranslator;

    private int size;

    private EventHandler handler;

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    public RingBufferComponent(int size, EventHandler handler) {
        this.size = size;
        this.handler = handler;
    }


    public void start() {
        EventFactory<TcpMessageEvent> eventFactory = new TcpMessageEventFactory();
        tcpMessageEventTranslator = new TcpMessageEventTranslator();
        disruptor = new Disruptor<>(eventFactory, this.size, new TcpMessageThreadFactory(), ProducerType.MULTI, new BlockingWaitStrategy());
        disruptor.handleEventsWith(this.handler);
        ringBuffer = disruptor.start();
        logger.info("ringBuffer 启动成功");
    }

    public void reset() {
        try {
            this.shutdown();
        } catch (Exception e) {
            logger.error("reset error", e);
        }
        this.start();
    }

    public void shutdown() {
        disruptor.shutdown();
    }

    public void pushData(TcpMessage message) {
        ringBuffer.publishEvent(tcpMessageEventTranslator, message);
    }

}
